package com.cn.client;

import com.cn.client.impl.DefaultMQPushConsumerImpl;
import com.cn.common.MessageQueue;
import com.cn.common.ProcessQueue;
import com.cn.common.PullRequest;
import com.cn.common.consumer.MessageQueueAllLock;
import com.cn.exception.MyException;
import com.cn.util.UtilAll;

import java.util.*;
import java.util.concurrent.*;

public class RemplanceImpl {

    private DefaultMQPushConsumerImpl defaultMQPushConsumer;
    private MQClientInstance mqClientInstance;

    public ConcurrentMap<String /*topic*/, Set<MessageQueue>> messageQueueTables = new ConcurrentHashMap<>();
    public ConcurrentMap<MessageQueue, ProcessQueue> processQueueTables = new ConcurrentHashMap<>();
    private ScheduledExecutorService scheduledExecutorService;

    public RemplanceImpl(DefaultMQPushConsumerImpl defaultMQPushConsumer, MQClientInstance mqClientInstance) {
        this.defaultMQPushConsumer = defaultMQPushConsumer;
        this.mqClientInstance = mqClientInstance;
        scheduledExecutorService = Executors.newScheduledThreadPool(8);
    }

    /**
     * 锁定队列
     *
     * @param messageQueue
     */
    public boolean lock(MessageQueue messageQueue) {
        MessageQueueAllLock messageQueueAllLock = new MessageQueueAllLock();

        Set<MessageQueue> reqMessageQueue = new HashSet<>();
        reqMessageQueue.add(messageQueue);

        messageQueueAllLock.setGroup(this.defaultMQPushConsumer.getConsumeGroup());
        messageQueueAllLock.setClientId(this.mqClientInstance.getClientId());
        messageQueueAllLock.setMessageQueueSet(reqMessageQueue);

        try {
            //服务器返回的锁定队列列表
            Set<MessageQueue> result = this.mqClientInstance.getMqClientApi().lockAllQueue(messageQueueAllLock, messageQueue.getBrokerName());
//            System.out.println("broker端锁定队列：" + result);
//            System.out.println("需要锁定的队列：" + messageQueue);
            if (result != null && !result.isEmpty()) {
                for (MessageQueue mq : result) {
                    //如果是第一次启动，那么以下代码不执行
                    ProcessQueue processQueue = processQueueTables.get(mq);
                    if (processQueue != null) {
                        processQueue.setLocked(true);
                    }
                }
                return result.contains(messageQueue);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 锁定所有队列
     */
    public void lockAll() {
//        System.out.println("正在执行锁定任务" + this.mqClientInstance.getClientId());
        ConcurrentMap<String, Set<MessageQueue>> messageTables = buildMessageQueue();
        for (String key : messageTables.keySet()) {
            MessageQueueAllLock messageQueueAllLock = new MessageQueueAllLock();
            messageQueueAllLock.setGroup(this.defaultMQPushConsumer.getConsumeGroup());
            messageQueueAllLock.setClientId(this.mqClientInstance.getClientId());

            //查询该topic下的锁定队列
            Set<MessageQueue> reqMessageQueue = messageTables.get(key);
            messageQueueAllLock.setMessageQueueSet(reqMessageQueue);

            try {
                Set<MessageQueue> result = this.mqClientInstance.getMqClientApi().lockAllQueue(messageQueueAllLock, key);
//                System.out.println("result:" + result);
                if (result != null) {
                    for (MessageQueue mq : result) {
                        ProcessQueue processQueue = processQueueTables.get(mq);
//                        System.out.println("==>" + processQueue);
                        if (processQueue != null) {
                            processQueue.setLocked(true);//已经在服务端锁定了
                        }
                    }

                    for (MessageQueue mq : reqMessageQueue) {
                        if (!result.contains(mq)) {//锁定队列中没有当前节点
                            ProcessQueue processQueue = processQueueTables.get(mq);
                            if (processQueue != null) {
//                                System.out.println("++>");
                                processQueue.setLocked(false);
                            }
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 解锁
     */
    public boolean unLockAll(MessageQueue mq, ProcessQueue pq) {
        //1.持久化该队列的位移
        this.defaultMQPushConsumer.getOffsetStore().persit(mq);
        //2.移除
        this.defaultMQPushConsumer.getOffsetStore().remove(mq);
        try {
            //锁定线程（因为可能消息正在消费过程中，因此必须保证消费完毕后才可以移除）
            if (pq.getLockConsume().tryLock(1000, TimeUnit.MILLISECONDS)) {
                try {
                    //从服务端移除锁定队列
                    removepq(mq, pq);
                    return true;
                } catch (Exception e) {

                } finally {
                    pq.getLockConsume().unlock();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public void removepq(MessageQueue mq, ProcessQueue pq) {
        //判断是否存在正在处理的消息
        if (pq.hasTempMsg()) {
            scheduledExecutorService.schedule(new Runnable() {
                @Override
                public void run() {
                    unLock(mq);
                }
            }, 20000, TimeUnit.MILLISECONDS);
        } else {
            unLock(mq);
        }
    }

    public void unLock(MessageQueue mq) {
        this.mqClientInstance.getMqClientApi().unLockQueue(mq, this.defaultMQPushConsumer.getConsumeGroup(), this.mqClientInstance.getClientId());
    }


    public void doRemplance() {
        for (String key : this.messageQueueTables.keySet()) {
            //该topic所有的队列集
            Set<MessageQueue> messageQueueSetList = this.messageQueueTables.get(key);

            //处理负载均衡，当前消费节点可以消费那些队列
            List<String> cidAll = this.mqClientInstance.pullClientList(key);
//            System.out.println("服务器端已注册消费者列表：" + cidAll);
            if (cidAll == null || cidAll.isEmpty()) {
//                System.out.println(RemplanceImpl.class.getName() + "消费者节点没有注册：" + cidAll);
                continue;
            }
            List<MessageQueue> messageQueueList = new ArrayList<>();
            messageQueueList.addAll(messageQueueSetList);
            //排序
            Collections.sort(messageQueueList);
            Collections.sort(cidAll);
//            System.out.println("队列：" + messageQueueList);
            AllocateMessageQueueStrategy allocateMessageQueueStrategy = this.defaultMQPushConsumer.getAllocateMessageQueueStrategy();
            List<MessageQueue> result = new ArrayList<>();
            try {
                result = allocateMessageQueueStrategy.allocate(this.defaultMQPushConsumer.getConsumeGroup(), messageQueueList, cidAll, this.mqClientInstance.getClientId());
//                System.out.println("-->分配后：" + result);
                updateProcessQueueTableInRebalance(this.defaultMQPushConsumer.getConsumeGroup(), key, result);
            } catch (MyException e) {
                e.printStackTrace();
            }

        }
    }

    public void updateProcessQueueTableInRebalance(String group, String topic, List<MessageQueue> messageQueueSetList) {
        if (messageQueueSetList.isEmpty()) {
            return;
        }
        //处理旧队列
        Iterator<Map.Entry<MessageQueue, ProcessQueue>> iterator = this.processQueueTables.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<MessageQueue, ProcessQueue> entry = iterator.next();
            MessageQueue mq = entry.getKey();
            ProcessQueue pq = entry.getValue();
            if (mq.getTopic().equals(topic)) {
                //新锁定的列表中没有当前队列
                if (!messageQueueSetList.contains(mq)) {
                    //移除
                    pq.setDropped(true);
                    if (unLockAll(mq, pq)) {
                        iterator.remove();
                    }
                } else {
                    if (pq.isPullExpired()) {
                        //移除
                        pq.setDropped(true);
                        if (unLockAll(mq, pq)) {
                            iterator.remove();
                        }
                    }
                }
            }
        }
//        System.out.println("MQ队列长度：" + messageQueueSetList);
        List<PullRequest> pullRequestList = new ArrayList<>();
        //处理新队列
        for (MessageQueue mq : messageQueueSetList) {
            if (!this.processQueueTables.containsKey(mq)) {
                System.out.println("1.->");
                if (!lock(mq)) {
                    System.out.println(mq.toString() + "：锁定队列失败！");
                    continue;
                }
//                System.out.println("2.->");
                ProcessQueue processQueue = new ProcessQueue();
                long offset = defaultMQPushConsumer.getOffsetStore().readOffset(mq);
//                System.out.println("3.->");
                ProcessQueue pq = this.processQueueTables.putIfAbsent(mq, processQueue);
                if (pq != null) {

                } else {
                    PullRequest pullRequest = new PullRequest();
                    pullRequest.setConsumerGroup(group);
                    pullRequest.setMessageQueue(mq);
                    pullRequest.setProcessQueue(processQueue);
                    pullRequest.setNextOffset(offset);
                    pullRequestList.add(pullRequest);
                }
            } else {
                //该队列已经存在了
//                System.out.println("队列已经存在了，不需要重复执行");
            }

        }
//        System.out.println(RemplanceImpl.class.getName() + ":" + this.processQueueTables);
//        System.out.println("4.->");
        this.defaultMQPushConsumer.dispatchPullRequest(pullRequestList);
    }

    public ConcurrentMap<String, Set<MessageQueue>> buildMessageQueue() {
        ConcurrentMap<String, Set<MessageQueue>> queueTables = new ConcurrentHashMap<>();
        for (MessageQueue messageQueue : this.processQueueTables.keySet()) {
            Set<MessageQueue> set1 = queueTables.get(messageQueue.getBrokerName());
            if (set1 == null) {
                set1 = new HashSet<>();
                queueTables.put(messageQueue.getBrokerName(), set1);
            }
            set1.add(messageQueue);
        }
        return queueTables;
    }

}
